use std::io;

use log::{error, info};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpStream,
    time::{sleep, timeout, Duration},
};

use crate::{cli::Cli, error::Error};

pub async fn consume(mut stream: TcpStream, cli: &Cli) -> Result<(), Error> {
    let consume_command = format!("{} --topic {}\n", cli.command, cli.topic);
    match stream.try_write(consume_command.as_bytes()) {
        Ok(_) => {
            info!("command: {consume_command}");
        }
        Err(e) => {
            return Err(e.into());
        }
    }

    loop {
        let pull_command = format!("pull --topic {}\n", cli.topic);
        match stream.write_all(pull_command.as_bytes()).await {
            Ok(_m) => {
                info!("command: {pull_command}");
            }
            Err(e) => {
                error!("{}", e);
                break;
            }
        }

        stream.flush().await?;

        sleep(Duration::from_millis(200)).await;

        let read_timeout = Duration::from_millis(200);

        // 使用 timeout 宏尝试读取数据
        let result = timeout(read_timeout, async {
            let mut buffer = [0; 1024 * 100];
            match stream.read(&mut buffer).await {
                Ok(n) if n > 0 => Ok(Some(String::from_utf8_lossy(&buffer[..n]).to_string())),
                Ok(_) => Ok(None),
                Err(e) => Err(e),
            }
        })
        .await;

        match result {
            Ok(Ok(Some(response))) => {
                println!("Received: \n{}", response);
            }
            Ok(Ok(None)) => {
                // 没有数据到达，跳过处理
                println!("No data received, skipping.");
            }
            Ok(Err(e)) => {
                eprintln!("Failed to read data: {}", e);
                continue;
            }
            Err(_) => {
                eprintln!("Read operation timed out.");
                // 可选: 根据需要决定是否继续等待
                // break;
            }
        }

        sleep(Duration::from_millis(500)).await;
    }


    Ok(())
}
